package org.gxlu.com.jstorm.day1;

 
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class WCTopologyMain {

	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("wcspout", new WCSpount(),1);
		builder.setBolt("splitbolt", new SplitBolt(),2).localOrShuffleGrouping("wcspout") ;
		builder.setBolt("countbolt", new CountBolt(),2).fieldsGrouping("splitbolt", new Fields("word")) ;
		Config config = new Config();
		
		config.setNumWorkers(2);
		//config.setNumAckers(1);
		config.setDebug(true);
		
		
		//Map map = new HashMap();
		//map.put(Config.TOPOLOGY_WORKERS, 2);
		//map.put(Config.STORM_CLUSTER_MODE, "distributed");
		//Config.setNumAckers(map, 2);
	 
		if(args ==null || args.length ==0){
			LocalCluster local = new LocalCluster();
			local.submitTopology("topt01", config, builder.createTopology());
		}else{
			try {
				StormSubmitter.submitTopology(args[0],config, builder.createTopology());
			} catch (AlreadyAliveException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		 
				
				
	}

}
